package a_aa_amainpackage.lvyou_clear.lv_update.util

import com.alibaba.fastjson.JSON
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.elasticsearch.spark._

/*
*项目名: o2odata_commend
*文件名: data_fromes
*创建者: hancongcong
*创建时间:2019/12/24 10:54
*描述: TODO
*/ object jing_bianqian {

  val platform = "tuniu"

  /**
    * 更新地址
    *
    * @param sc
    * @param sqlContext
    * @param year
    * @param month
    * @param plat_type
    */
  def updateaddress(frame: DataFrame, sc: SparkContext, sqlContext: SQLContext, year: String, month: String, plat_type: String): DataFrame = {
    //    for (plat_type <- Array("ly")) {
    //      for (year <- Array("2020")) {
    //        //"1","2","3","4","5","6","7","8","9","10","11","12"
    //        for (month <- Array("3")) {
    //          println("updateaddress：：：：month:" + s"${month}")
    //          2019_update_tuniu_mp/tuniu_mp_2019_11
    frame.registerTempTable("t1")
    //where toursim_tickets_type = '03002'
    val data = sqlContext.sql(
      """
        |select * from t1
        |""".stripMargin)
    /*sc.esJsonRDD(s"${year}_update_${platform}_${plat_type}/${platform}_${plat_type}_${year}_${month}",
            """
              |{
              |  "size": 0
              |  , "query": {
              |    "term": {
              |      "toursim_tickets_type.keyword": {
              |        "value": "03002"
              |      }
              |    }
              |  },"aggs": {
              |    "NAME": {
              |      "terms": {
              |        "field": "first_arriving.keyword",
              |        "size": 500
              |      }
              |    }
              |  }
              |}
            """.stripMargin).values*/
    //          sqlContext.read.json(data).repartition(1).write.json(s"s3a://o2o-dataproces-group/zsc/platform/lvyou_lyv1/${plat_type}/${platform}/${year}/${month}")

    val result = update_Address(data, sqlContext)
    result
      .select("province_out", "city_out", "region_id", "economic_division_out").distinct()
      .show()
    //          toes(result, year, month, plat_type)
    result.registerTempTable("t2")
    //处理代码
    //where city_out = '0' and province_out != '0'
    val data1 = sqlContext.sql(
      """
        |select * from t2
        |""".stripMargin)

//      sc.esJsonRDD(s"${year}_update_${platform}_${plat_type}/${platform}_${plat_type}_${year}_${month}",
//      """
//        |{
//        |  "query": {
//        |    "bool": {
//        |      "must": [
//        |        {"match": {
//        |          "city_out.keyword": "0"
//        |        }}
//        |      ],
//        |      "must_not": [
//        |        {"match": {
//        |          "province_out.keyword": "0"
//        |        }}
//        |      ]
//        |    }
//        |  }
//        |}
//            """.stripMargin).values
    val result1 = update_Address(data1, sqlContext)
//    toes(result1, year, month, plat_type)
    //        }
    //      }
    //    }
    result1
  }


  def toes(data: DataFrame, year: String, month: String, plat_type: String): Unit = {
    val value = data.toJSON.rdd.map(line => {
      val lines = JSON.parseObject(line)
      lines
    })
    //    value.saveToEs(s"${year}_update_${platform}_${plat_type}/${platform}_${plat_type}_${year}_${month}18", Map("es.mapping.id" -> "product_Id"))
    value.saveToEs(s"${year}_update_${platform}_${plat_type}/${platform}_${plat_type}_${year}_${month}", Map("es.mapping.id" -> "product_Id"))
  }

  def update_Address(data: DataFrame, sqlContext: SQLContext): DataFrame = {

    data.registerTempTable("data")

    //修改目的地省份和城市
    val provinceAndCity_data = sqlContext.sql(
      """
        |select
        |*,
        |case
        |--匹配目的地城市含族的城市
        |when first_arriving rlike '延边' then '延边朝鲜族自治州'
        |when first_arriving rlike '大理' then '大理白族自治州'
        |when first_arriving rlike '丽江' then '丽江市'
        |when first_arriving rlike '秀山' then '秀山土家族苗族自治县'
        |when first_arriving rlike '迪庆' then '迪庆藏族自治州'
        |when first_arriving rlike '湘西' then '湘西土家族苗族自治州'
        |when first_arriving rlike '石柱' then '石柱土家族自治县'
        |when first_arriving rlike '临夏' then '临夏回族自治州'
        |when first_arriving rlike '凉山' then '凉山彝族自治州'
        |when first_arriving rlike '恩施' then '恩施土家族苗族自治州'
        |when first_arriving rlike '海北' then '海北藏族自治州'
        |when first_arriving rlike '红河' then '红河哈尼族彝族自治州'
        |when first_arriving rlike '海南' then '三亚市'
        |when first_arriving rlike '文山' then '文山壮族苗族自治州'
        |when first_arriving rlike '怒江' then '怒江傈僳族自治州'
        |when first_arriving rlike '阿坝' then '阿坝藏族羌族自治州'
        |when first_arriving rlike '果洛' then '果洛藏族自治州'
        |when first_arriving rlike '西双版纳' then '西双版纳傣族自治州'
        |when first_arriving rlike '彭水' then '彭水苗族土家族自治县'
        |when first_arriving rlike '楚雄' then '楚雄彝族自治州'
        |when first_arriving rlike '甘南' then '甘南藏族自治州'
        |when first_arriving rlike '海西' then '海西蒙古族藏族自治州'
        |when first_arriving rlike '黄南' then '黄南藏族自治州'
        |when first_arriving rlike '酉阳' then '酉阳土家族苗族自治县'
        |when first_arriving rlike '玉树' then '玉树藏族自治州'
        |when first_arriving rlike '德宏' then '德宏傣族景颇族自治州'
        |when first_arriving rlike '黔南' then '黔南布依族苗族自治州'
        |when first_arriving rlike '昌吉' then '昌吉回族自治州'
        |when first_arriving rlike '黔西' then '黔西南布依族苗族自治州'
        |when first_arriving rlike '黔东' then '黔东南苗族侗族自治州'
        |when first_arriving rlike '甘孜' then '甘孜藏族自治州'
        |
        |when first_arriving rlike '锡林郭勒' then '锡林郭勒盟'
        |
        |
        |when first_arriving rlike '阿克苏'      then     '阿克苏地区'
        |when first_arriving rlike '喀什'      then   '喀什地区'
        |when first_arriving rlike '和田'      then   '和田地区'
        |when first_arriving rlike '塔城'      then   '塔城地区'
        |when first_arriving rlike '大兴安岭'      then   '大兴安岭地区'
        |when first_arriving rlike '阿勒泰'      then   '阿勒泰地区'
        |when first_arriving rlike '那曲'      then   '那曲地区'
        |when first_arriving rlike '阿里'      then       '阿里地区'
        |when first_arriving rlike '神农架'      then       '十堰市'
        |when first_arriving rlike '兴安'      then       '兴安盟'
        |
        |
        |
        |
        |--匹配目的地出现省份的城市
        |when first_arriving rlike '甘肃' then   '0'
        |when first_arriving rlike '黑龙江' then     '0'
        |when first_arriving rlike '江西' then   '0'
        |when first_arriving rlike '辽宁' then   '0'
        |when first_arriving rlike '浙江' then   '杭州市'
        |when first_arriving rlike '西藏' then    '0'
        |when first_arriving rlike '上海' then   '0'
        |when first_arriving rlike '云南' then   '丽江市'
        |when first_arriving rlike '河南' then   '0'
        |when first_arriving rlike '江苏' then   '0'
        |when first_arriving rlike '天津' then   '0'
        |when first_arriving rlike '广西' then    '0'
        |when first_arriving rlike '山西' then   '0'
        |when first_arriving rlike '广东' then   '广州市'
        |when first_arriving rlike '北京' then   '0'
        |when first_arriving rlike '吉林' then   '0'
        |when first_arriving rlike '湖南' then   '0'
        |when first_arriving rlike '湖北' then   '0'
        |when first_arriving rlike '重庆' then   '0'
        |when first_arriving rlike '福建' then   '0'
        |when first_arriving rlike '海南' then   '0'
        |when first_arriving rlike '台湾' then   '0'
        |when first_arriving rlike '新疆' then    '乌鲁木齐市'
        |when first_arriving rlike '安徽' then   '0'
        |when first_arriving rlike '内蒙古' then    '0'
        |when first_arriving rlike '山东' then   '0'
        |when first_arriving rlike '陕西' then   '0'
        |when first_arriving rlike '宁夏' then    '银川市'
        |when first_arriving rlike '青海' then   '西宁市'
        |when first_arriving rlike '四川' then   '成都市'
        |when first_arriving rlike '河北' then   '0'
        |when first_arriving rlike '贵州' then   '0'
        |when first_arriving rlike '天目湖'      then       '常州市'
        |when first_arriving rlike '中华恐龙园'      then       '常州市'
        |when first_arriving rlike '千岛湖'      then       '杭州市'
        |when first_arriving rlike '巴音郭楞'      then       '巴音郭楞蒙古自治州'
        |when first_arriving rlike '密云区'      then       '北京市'
        |when first_arriving rlike '巢湖市'      then       '巢湖市'
        |when first_arriving rlike '成都'      then       '成都市'
        |when first_arriving rlike '溧阳市'      then       '常州市'
        |when first_arriving rlike '莫干山'      then       '湖州市'
        |when first_arriving rlike '三亚'      then       '三亚市'
        |when first_arriving rlike '三英温泉'      then       '广州市'
        |else city_out end as city_out_v1,
        |case
        |--匹配目的地出现省份的
        |when first_arriving rlike '甘肃' then '甘肃省'
        |when first_arriving rlike '黑龙江' then  '黑龙江省'
        |when first_arriving rlike '江西' then '江西省'
        |when first_arriving rlike '辽宁' then '辽宁省'
        |when first_arriving rlike '浙江' then '浙江省'
        |when first_arriving rlike '西藏' then '西藏自治区'
        |when first_arriving rlike '上海' then '上海市'
        |when first_arriving rlike '云南' then '云南省'
        |when first_arriving rlike '河南' then '河南省'
        |when first_arriving rlike '江苏' then '江苏省'
        |when first_arriving rlike '天津' then '天津市'
        |when first_arriving rlike '广西' then '广西壮族自治区'
        |when first_arriving rlike '山西' then '山西省'
        |when first_arriving rlike '广东' then '广东省'
        |when first_arriving rlike '北京' then '北京市'
        |when first_arriving rlike '吉林' then '吉林省'
        |when first_arriving rlike '湖南' then '湖南省'
        |when first_arriving rlike '湖北' then '湖北省'
        |when first_arriving rlike '重庆' then '重庆市'
        |when first_arriving rlike '福建' then '福建省'
        |when first_arriving rlike '海南' then '海南省'
        |when first_arriving rlike '新疆' then '新疆维吾尔自治区'
        |when first_arriving rlike '安徽' then '安徽省'
        |when first_arriving rlike '内蒙古' then '内蒙古自治区'
        |when first_arriving rlike '山东' then '山东省'
        |when first_arriving rlike '陕西' then '陕西省'
        |when first_arriving rlike '宁夏' then '宁夏回族自治区'
        |when first_arriving rlike '青海' then '青海省'
        |when first_arriving rlike '四川' then '四川省'
        |when first_arriving rlike '河北' then '河北省'
        |when first_arriving rlike '贵州' then '贵州省'
        |--匹配目的地城市含族的省份
        |when first_arriving rlike '延边' then         '吉林省'
        |when first_arriving rlike '秀山' then         '重庆市'
        |when first_arriving rlike '迪庆' then         '云南省'
        |when first_arriving rlike '湘西' then         '湖南省'
        |when first_arriving rlike '石柱' then         '重庆市'
        |when first_arriving rlike '临夏' then         '甘肃省'
        |when first_arriving rlike '凉山' then         '四川省'
        |when first_arriving rlike '恩施' then         '湖北省'
        |when first_arriving rlike '海北' then         '青海省'
        |when first_arriving rlike '红河' then         '云南省'
        |when first_arriving rlike '文山' then         '云南省'
        |when first_arriving rlike '怒江' then         '云南省'
        |when first_arriving rlike '阿坝' then         '四川省'
        |when first_arriving rlike '果洛' then         '青海省'
        |when first_arriving rlike '西双版纳' then         '云南省'
        |when first_arriving rlike '彭水' then         '重庆市'
        |when first_arriving rlike '楚雄' then         '云南省'
        |when first_arriving rlike '甘南' then         '甘肃省'
        |when first_arriving rlike '海西' then         '青海省'
        |when first_arriving rlike '黄南' then         '青海省'
        |when first_arriving rlike '酉阳' then         '重庆市'
        |when first_arriving rlike '玉树' then         '青海省'
        |when first_arriving rlike '德宏' then         '云南省'
        |when first_arriving rlike '黔南' then         '贵州省'
        |when first_arriving rlike '昌吉' then         '新疆维吾尔自治区'
        |when first_arriving rlike '黔西' then         '贵州省'
        |when first_arriving rlike '黔东' then         '贵州省'
        |when first_arriving rlike '甘孜' then         '四川省'
        |
        |when first_arriving rlike '阿克苏'      then     '新疆维吾尔自治区'
        |when first_arriving rlike '喀什'      then   '新疆维吾尔自治区'
        |when first_arriving rlike '和田'      then   '新疆维吾尔自治区'
        |when first_arriving rlike '塔城'      then   '新疆维吾尔自治区'
        |when first_arriving rlike '大兴安岭'      then   '黑龙江省'
        |when first_arriving rlike '阿勒泰'      then   '新疆维吾尔自治区'
        |when first_arriving rlike '那曲'      then   '西藏自治区'
        |when first_arriving rlike '阿里'      then       '西藏自治区'
        |when first_arriving rlike '兴安'      then       '内蒙古自治区'
        |when first_arriving rlike '锡林郭勒' then '内蒙古自治区'
        |when first_arriving rlike '大理' then '云南省'
        |when first_arriving rlike '丽江' then '云南省'
        |when first_arriving rlike '神农架'      then       '湖北省'
        |when first_arriving rlike '天目湖'      then       '江苏省'
        |when first_arriving rlike '中华恐龙园'      then       '江苏省'
        |when first_arriving rlike '千岛湖'      then       '浙江省'
        |when first_arriving rlike '巴音郭楞'      then       '新疆维吾尔自治区'
        |when first_arriving rlike '密云区'      then       '北京市'
        |when first_arriving rlike '巢湖市'      then       '安徽省'
        |when first_arriving rlike '成都'      then       '四川省'
        |when first_arriving rlike '溧阳市'      then       '江苏省'
        |when first_arriving rlike '莫干山'      then       '浙江省'
        |when first_arriving rlike '三亚'      then       '海南省'
        |when first_arriving rlike '三英温泉'      then       '广东省'
        |
        |
        |else province_out end as province_out_v1
        |from
        |data
      """.stripMargin).drop("city_out", "province_out")
      .withColumnRenamed("city_out_v1", "city_out")
      .withColumnRenamed("province_out_v1", "province_out")
    //    provinceAndCity_data.selectExpr("province_out").where("city_out='0'").show()
    //修改区域id 四大区
    provinceAndCity_data.registerTempTable("provinceAndCity_data")
    //标准地址表
    //    val address = sqlContext.read.option("header","true")
    //      .csv("s3a://o2o-dataproces-group/han_congcong/lv_address/")
    val address = sqlContext.read.option("header", "true")
      .csv("s3a://o2o-dimension-table/address_table/address_table_2019/address_mapping_standard/2019_9/")
      .where("district='0'")
    address.createOrReplaceTempView("address")
    val regionidAndEconomic = sqlContext.sql(
      """
        |select
        |t1.*,
        |case when t2.city is null then '0' else t2.uniqueid end as region_id_v1,
        |case when t2.city is null then '0' else t2.economic_division end as economic_division_out_v1
        |from
        |provinceAndCity_data t1
        |left join
        |address t2
        |on t1.province_out=t2.province and t1.city_out=t2.city
      """.stripMargin).drop("region_id", "economic_division_out")
      .withColumnRenamed("region_id_v1", "region_id")
      .withColumnRenamed("economic_division_out_v1", "economic_division_out")
    regionidAndEconomic.registerTempTable("regionidAndEconomic")
    //修改toursim_tickets_type
    val toursim_tickets_typedata = sqlContext.sql(
      """
        |select
        |*,
        |case when province_out in ("湖南省"
        |,"甘肃省"
        |,"新疆维吾尔自治区"
        |,"安徽省"
        |,"贵州省"
        |,"河南省"
        |,"云南省"
        |,"北京市"
        |,"辽宁省"
        |,"浙江省"
        |,"西藏自治区"
        |,"上海市"
        |,"重庆市"
        |,"青海省"
        |,"海南省"
        |,"宁夏回族自治区"
        |,"山西省"
        |,"湖北省"
        |,"江西省"
        |,"四川省"
        |,"广东省"
        |,"山东省"
        |,"内蒙古自治区"
        |,"江苏省"
        |,"福建省"
        |,"天津市"
        |,"广西壮族自治区"
        |,"陕西省"
        |,"黑龙江省"
        |,"吉林省"
        |,"河北省") then '03001' else '03002' end as toursim_tickets_type_v1
        |from
        |regionidAndEconomic
      """.stripMargin).drop("toursim_tickets_type")
      .withColumnRenamed("toursim_tickets_type_v1", "toursim_tickets_type")
    //补全6位id,更新四大区
    val newregionidAndEconomic = toursim_tickets_typedata.toJSON.rdd.map(line => {
      val json = JSON.parseObject(line)
      val regid_info = json.getOrDefault("region_id_info", "-1").toString
      val region_id = json.getOrDefault("region_id", "-1").toString
      val district_out = json.getOrDefault("district_out", "0").toString
      json.put("district_out", district_out)
      val economic_division_out = json.getOrDefault("economic_division_out", "-1").toString
      val province_out = json.getOrDefault("province_out", "-1").toString

      var region_id_info = "0"
      var region_id_out = "0"

      if (regid_info.length == 2) {
        region_id_info = regid_info + "0000"
        json.put("region_id_info", region_id_info)
      } else if (regid_info.length == 4) {
        region_id_info = regid_info + "00"
        json.put("region_id_info", region_id_info)
      } else if (regid_info.length == 6) {
        region_id_info = regid_info
        json.put("region_id_info", region_id_info)
      } else {
        region_id_info = "0"
        json.put("region_id_info", region_id_info)
      }


      if (region_id.length == 2) {
        region_id_out = region_id + "0000"
        json.put("region_id", region_id_out)
      } else if (region_id.length == 4) {
        region_id_out = region_id + "00"
        json.put("region_id", region_id_out)
      } else if (region_id.length == 6) {
        region_id_out = region_id
        json.put("region_id", region_id_out)
      } else {
        region_id_out = "0"
        json.put("region_id", region_id_out)
      }


      if (economic_division_out.equals("-1")) {
        json.put("economic_division_out", "0")
      } else if (economic_division_out.equals("西部地区")) {
        json.put("economic_division_out", "4")
      } else if (economic_division_out.equals("东部地区")) {
        json.put("economic_division_out", "2")
      } else if (economic_division_out.equals("中部地区")) {
        json.put("economic_division_out", "3")
      } else if (economic_division_out.equals("东北地区")) {
        json.put("economic_division_out", "1")
      } else if (economic_division_out.equals("港澳") || economic_division_out.equals("台湾")) {
        if (province_out.contains("香港")) {
          json.put("economic_division_out", "香港特别行政区")
        } else if (province_out.contains("澳门")) {
          json.put("economic_division_out", "澳门特别行政区")
        } else if (province_out.contains("台湾")) {
          json.put("economic_division_out", "台湾省")
        }
      }
      json.toJSONString
    })
    sqlContext.read.json(newregionidAndEconomic)
  }

}
